package com.shop.zkmall.app.service.a.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.shop.zkmall.app.common.RocketConstants;
import com.shop.zkmall.app.controller.a.live.LiveUrlUtil;
import com.shop.zkmall.app.controller.a.live.TxPlayUrlUtil;
import com.shop.zkmall.app.dao.buyer.CereBuyerUserDAO;
import com.shop.zkmall.app.domain.*;
import com.shop.zkmall.app.domain.req.*;
import com.shop.zkmall.app.domain.res.PlayUrlVO;
import com.shop.zkmall.app.domain.res.ProductInfo;
import com.shop.zkmall.app.domain.res.RecommendRes;
import com.shop.zkmall.app.domain.res.StartLiveRes;
import com.shop.zkmall.app.im.TxTemplate;
import com.shop.zkmall.app.mapper.*;
import com.shop.zkmall.app.redis.service.api.impl.RedisService;
import com.shop.zkmall.app.service.a.ICereBuyerLiveRoomService;
import com.shop.zkmall.app.utils.LoginHelper;
import com.shop.zkmall.app.utils.TencentCloudEmailUtil;
import com.shop.zkmall.commons.domain.buyer.CereBuyerUser;
import com.shop.zkmall.commons.domain.common.PageParam;
import com.shop.zkmall.commons.result.R;
import jakarta.annotation.Resource;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.checkerframework.checker.units.qual.A;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import javax.security.auth.login.LoginContext;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.logging.Logger;

import static com.shop.zkmall.app.common.CommonRedisKey.LIVE_AUCTION_LIST;
import static com.shop.zkmall.app.controller.a.live.LiveConstant.PUSH_DOMAIN;
import static com.shop.zkmall.app.controller.a.live.LiveConstant.TX_SECRET;

/**
 * <p>
 * 直播间信息表 服务实现类
 * </p>
 *
 * @author ${author}
 * @since 2025-11-25
 */
@Service
public class CereBuyerLiveRoomServiceImpl extends ServiceImpl<CereBuyerLiveRoomMapper, CereBuyerLiveRoom> implements ICereBuyerLiveRoomService {



    @Autowired
    private CereBuyerAdPromotionMapper cereBuyerAdPromotionMapper;
    @Autowired
    private CereBuyerUserFollowMapper cereBuyerUserFollowMapper;
    @Autowired
    private CereBuyerLiveCategoryMapper cereBuyerLiveCategoryMapper;
    @Autowired
    private CereBuyerUserDAO cereBuyerUserDAO;
    @Autowired
    private CereBuyerGiftRecordServiceImpl cereBuyerGiftRecordService;
    @Autowired
    private CereLiveJoinRecordMapper cereLiveJoinRecordMapper;
    @Autowired
    private CereShopMarketingMapper cereShopMarketingMapper;
    @Autowired
    private LiveRoomGoodsMapper liveRoomGoodsMapper;
    @Autowired
    private RedisService redisService;
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @Autowired
    private TxTemplate txTemplate;
    @Resource(name = "commonExecutor")
    private ThreadPoolTaskExecutor executor;



    @Override
    public List<RecommendRes> recommendList(PageParam param) {

        //查询出参与广告推广的主播
        List<CereBuyerLiveRoom> cereBuyerLiveRooms = getAdPromotionAnchor();

        List<RecommendRes> resData = new ArrayList<>();
        if (!CollUtil.isEmpty(cereBuyerLiveRooms)){
            List<RecommendRes> res = buildLiveRoomRes(cereBuyerLiveRooms);
            resData.addAll(res);
        }

        Page<CereBuyerLiveRoom> cereBuyerLiveRoomPage = new Page<>();
        cereBuyerLiveRoomPage.setCurrent(param.getPage());
        cereBuyerLiveRoomPage.setSize(param.getPageSize());
        Page<CereBuyerLiveRoom> page = lambdaQuery().page(cereBuyerLiveRoomPage);
        if (CollUtil.isNotEmpty(page.getRecords())){
            List<RecommendRes> res = buildLiveRoomRes(page.getRecords());
            if (CollUtil.isNotEmpty(resData)){
                if (resData.size() >= param.getPageSize()){
                    resData.addAll(res.subList(0, param.getPageSize()));
                    return resData;
                }else {
                    //不够就补齐
                    resData.addAll(res);
                    return resData;
                }
            }
        }
        //推荐主播
        return List.of();
    }

    @Override
    public List<RecommendRes> attentionList(PageParam param, CereBuyerUser user) {
        //查询我所关注的人
        LambdaQueryWrapper<CereBuyerUserFollow> cereBuyerUserFollowLambdaQueryWrapper = new LambdaQueryWrapper<>();
        cereBuyerUserFollowLambdaQueryWrapper.eq(CereBuyerUserFollow::getUserId, user.getBuyerUserId());
        List<CereBuyerUserFollow> cereBuyerUserFollows = cereBuyerUserFollowMapper.selectList(cereBuyerUserFollowLambdaQueryWrapper);
        if (CollUtil.isEmpty(cereBuyerUserFollows)){
            return List.of();
        }
        List<CereBuyerLiveRoom> list = lambdaQuery().in(CereBuyerLiveRoom::getAnchorId, cereBuyerUserFollows).list();
        List<RecommendRes> res = buildLiveRoomRes(list);
        return res;
    }

    @Override
    public List<RecommendRes> categoryList(LiveCateGoryReq param, CereBuyerUser user) {
        List<CereBuyerLiveRoom> list = lambdaQuery().eq(CereBuyerLiveRoom::getCategoryId, param.getCategoryId()).list();
        if (CollUtil.isEmpty(list)){
            return List.of();
        }
        List<RecommendRes> res = buildLiveRoomRes(list);
        return res;
    }

    @Override
    public List<RecommendRes> searchLiveList(LiveSearchReq param, CereBuyerUser user) {
        //查询主播昵称和直播标题相识的直播
        List<CereBuyerLiveRoom> list = lambdaQuery()
                .like(CereBuyerLiveRoom::getAnchorName, param.getKeyWord())
                .or().like(CereBuyerLiveRoom::getTitle, param.getKeyWord()).list();

        if (CollUtil.isEmpty(list)){
            return List.of();
        }

        List<RecommendRes> res = buildLiveRoomRes(list);

        //添加搜索记录 存入Redis

        return res;
    }

    @Override
    public R<?> saveLiveSetting(CereBuyerLiveRoomReq cereBuyerLiveRoomReq) {
        Long userId = LoginHelper.getUserId();
        CereBuyerUser cereBuyerUser = cereBuyerUserDAO.selectByPrimaryKey(userId);

        //设置直播间基本信息
        CereBuyerLiveRoom cereBuyerLiveRoom = new CereBuyerLiveRoom();
        cereBuyerLiveRoom.setAnchorId(cereBuyerUser.getBuyerUserId());
        cereBuyerLiveRoom.setAnchorName(cereBuyerUser.getName());
        cereBuyerLiveRoom.setAnchorAvatar(cereBuyerUser.getHeadImage());
        cereBuyerLiveRoom.setTitle(cereBuyerLiveRoomReq.getTitle());
        cereBuyerLiveRoom.setCoverUrl(cereBuyerLiveRoomReq.getCoverUrl());
        cereBuyerLiveRoom.setCategoryId(cereBuyerLiveRoomReq.getCategoryId());


        //设置分类名称
        LambdaQueryWrapper<CereBuyerLiveCategory> cereBuyerLiveCategoryLambdaQueryWrapper = new LambdaQueryWrapper<>();
        cereBuyerLiveCategoryLambdaQueryWrapper.eq(CereBuyerLiveCategory::getId, cereBuyerLiveRoomReq.getCategoryId());
        CereBuyerLiveCategory cereBuyerLiveCategory = cereBuyerLiveCategoryMapper.selectOne(cereBuyerLiveCategoryLambdaQueryWrapper);
        if (cereBuyerLiveCategory != null){
            cereBuyerLiveRoom.setCategoryName(cereBuyerLiveCategory.getCategoryName());
        }

        save(cereBuyerLiveRoom);

        return R.ok(cereBuyerLiveRoom);
    }

    @Override
    public R<?> startLive(StartLiveReq startLiveReq) {
        Long liveId = startLiveReq.getLiveRoomId();
        if (liveId == null) {
            return R.fail("直播间ID不能为空");
        }

        // 查询直播间信息
        CereBuyerLiveRoom room = getById(liveId);
        if (room == null) {
            return R.fail("直播间不存在");
        }

        // 已经在直播中
        if (room.getStatus() != null && room.getStatus() == 1) {
            return R.fail("该直播间已在直播中");
        }

        //创建群组
        String liveGroupId = txTemplate.createLiveGroup(room.getTitle(),
                room.getAnchorId().toString(), room.getTitle(), room.getCoverUrl());
        //加入群组 成为管理
        txTemplate.addAdminToLiveGroup(liveGroupId, room.getAnchorId().toString());

        //更新群聊id
        room.setGroupId(liveGroupId);

        // 生成 streamName（唯一）
        String streamName = "live_" + room.getAnchorId() + "_" + System.currentTimeMillis();

        // 生成推流地址
        String pushUrl = LiveUrlUtil.buildPushUrl(
                PUSH_DOMAIN,      // 推流域名
                streamName,
                TX_SECRET,        // 推流防盗链 key
                2 * 60 * 60       // 过期时间（2小时）
        );

        // 更新直播间信息
        room.setStatus(1);  // 1=直播中
        room.setStreamName(streamName);
        updateById(room);

        StartLiveRes startLiveRes = new StartLiveRes();
        startLiveRes.setPushUrl(pushUrl);
        startLiveRes.setStreamName(room.getStreamName());
        startLiveRes.setLiveRoomId(room.getId());
        return R.ok(startLiveRes);
    }

    @Override
    public R<?> getPlayUrl(Long roomId) {
        CereBuyerLiveRoom room = getById(roomId);

        if (room == null) {
            return R.fail("直播间不存在");
        }

        if (room.getStreamName() == null) {
            return R.fail("直播尚未开始，暂无播放流");
        }

        // 生成播放地址
        PlayUrlVO playUrlVO = TxPlayUrlUtil.buildPlayUrls(room.getStreamName());

        return R.ok(playUrlVO);
    }

    @Override
    public R<?> sendGift(SendGiftReq sendGiftReq) {
        CereBuyerLiveRoom byId = getById(sendGiftReq.getLiveRoomId());
        if (byId == null) {
            return R.fail("直播间不存在");
        }
        Long userId = LoginHelper.getUserId();
        //赠送礼物
        SendReq sendReq = new SendReq();
        sendReq.setSendGiftReq(sendGiftReq);
        sendReq.setUserId(userId);
        sendReq.setAnchorId(byId.getAnchorId());
        rocketMQTemplate.convertAndSend(RocketConstants.SEND_GIFT_TOPIC, sendReq);
        return R.ok();
    }

    @Override
    public R<?> searchLiveProduct(String keyWord) {
        List<CereBuyerLiveRoom> list = lambdaQuery().like(CereBuyerLiveRoom::getTitle, keyWord).list();
        return R.ok(list);
    }

    @Override
    public R<?> explainProduct(ExplainProductReq explainProductReq) {
        //将商品广播给所有观众
        LambdaQueryWrapper<LiveRoomGoods> liveRoomGoodsLambdaQueryWrapper = new LambdaQueryWrapper<>();
        liveRoomGoodsLambdaQueryWrapper.eq(LiveRoomGoods::getLiveRoomId, explainProductReq.getLiveRoomId())
                .eq(LiveRoomGoods::getGoodsId, explainProductReq.getProductId());
        LiveRoomGoods liveRoomGoods = liveRoomGoodsMapper.selectOne(liveRoomGoodsLambdaQueryWrapper);

        //查询商铺起拍价
        CereShopMarketing cereShopMarketing = cereShopMarketingMapper.selectById(liveRoomGoods.getGoodsId());
        if (liveRoomGoods != null){


            //免费抽奖商品
            if (liveRoomGoods.getType() == 0){
                ProductInfo productInfo = new ProductInfo();
                productInfo.setProductId(liveRoomGoods.getGoodsId().toString());
                productInfo.setTitle(liveRoomGoods.getGoodsName());
                productInfo.setImage(liveRoomGoods.getCoverImg());
                productInfo.setType(0);
                productInfo.setStock(liveRoomGoods.getStock());
                txTemplate.pushProductToLiveRoom(liveRoomGoods.getGoodsId().toString(), productInfo);
            }


            //砍价商品
            if (liveRoomGoods.getType() == 1){
                ProductInfo productInfo = new ProductInfo();
                productInfo.setProductId(liveRoomGoods.getGoodsId().toString());
                productInfo.setTitle(liveRoomGoods.getGoodsName());
                productInfo.setImage(liveRoomGoods.getCoverImg());
                productInfo.setPrice(cereShopMarketing.getOriginalPrice());
                productInfo.setStock(liveRoomGoods.getStock());
                productInfo.setType(1);
                productInfo.setCanOperate(0);
                txTemplate.pushProductToLiveRoom(liveRoomGoods.getGoodsId().toString(), productInfo);
            }



            //讲解竞拍商品
            if (liveRoomGoods.getType() == 2){
                ProductInfo productInfo = new ProductInfo();
                productInfo.setProductId(liveRoomGoods.getGoodsId().toString());
                productInfo.setTitle(liveRoomGoods.getGoodsName());
                productInfo.setImage(liveRoomGoods.getCoverImg());
                productInfo.setPrice(cereShopMarketing.getStartPrice());
                productInfo.setAddPrice(cereShopMarketing.getAddPrice());
                productInfo.setStock(liveRoomGoods.getStock());
                productInfo.setType(2);
                productInfo.setCanOperate(0);
                txTemplate.pushProductToLiveRoom(liveRoomGoods.getGoodsId().toString(), productInfo);
            }

            //闪购
            if (liveRoomGoods.getType() == 3){
                ProductInfo productInfo = new ProductInfo();
                productInfo.setProductId(liveRoomGoods.getGoodsId().toString());
                productInfo.setTitle(liveRoomGoods.getGoodsName());
                productInfo.setImage(liveRoomGoods.getCoverImg());
                productInfo.setPrice(cereShopMarketing.getOriginalPrice());
                productInfo.setStock(liveRoomGoods.getStock());
                productInfo.setType(3);
                productInfo.setCanOperate(0);
                txTemplate.pushProductToLiveRoom(liveRoomGoods.getGoodsId().toString(), productInfo);
            }





        }
        return R.ok();
    }

    @Override
    public synchronized R<?> joinLiveRoom(JoinLiveRoomReq joinLiveRoomReq) {
        //异步线程处理
        CompletableFuture.runAsync(() -> {
            Long userId = LoginHelper.getUserId();

            //调用im加入群组
            txTemplate.joinLiveGroup(joinLiveRoomReq.getLiveRoomId().toString(), userId.toString());

            //查询是否存在记录
            LambdaQueryWrapper<CereLiveJoinRecord> eq = new LambdaQueryWrapper<CereLiveJoinRecord>()
                    .eq(CereLiveJoinRecord::getUserId, userId)
                    .eq(CereLiveJoinRecord::getLiveId, joinLiveRoomReq.getLiveRoomId());
            List<CereLiveJoinRecord> cereLiveJoinRecords = cereLiveJoinRecordMapper.selectList(eq);
            if (CollUtil.isNotEmpty(cereLiveJoinRecords)){
                CereLiveJoinRecord last = CollUtil.getLast(cereLiveJoinRecords);
                last.setUpdateTime(DateUtil.toLocalDateTime(new Date()));
                cereLiveJoinRecordMapper.updateById(last);
            }else {
                CereLiveJoinRecord cereLiveJoinRecord = new CereLiveJoinRecord();
                cereLiveJoinRecord.setUserId(userId);
                cereLiveJoinRecord.setLiveId(joinLiveRoomReq.getLiveRoomId());
                cereLiveJoinRecord.setEnterTime(DateUtil.toLocalDateTime(new Date()));
                cereLiveJoinRecordMapper.insert(cereLiveJoinRecord);
            }
            log.warn("用户:" + userId + "-----加入直播间:" + joinLiveRoomReq.getLiveRoomId());

        }, executor);
        return R.ok();
    }

    @Override
    public R<?> liveAuctionSendPrice(LiveAuctionSendPriceReq liveAuctionSendPriceReq) {
        if (liveAuctionSendPriceReq.getType()==0){
            //自定义价格
            LiveRoomGoodsBid liveRoomGoodsBid = new LiveRoomGoodsBid();
            liveRoomGoodsBid.setRoomId(liveAuctionSendPriceReq.getLiveRoomId());
            liveRoomGoodsBid.setGoodsId(liveAuctionSendPriceReq.getProductId());
            liveRoomGoodsBid.setUserId(LoginHelper.getUserId());
            liveRoomGoodsBid.setBidAmount(liveAuctionSendPriceReq.getPrice());
            liveRoomGoodsBid.setBidTime(DateUtil.toLocalDateTime(new Date()));
            String key = "liveRoomGoodsBid:" + liveAuctionSendPriceReq.getLiveRoomId() + ":" + liveAuctionSendPriceReq.getProductId();
            if (redisService.hasKey(key)){
                String cacheObject = redisService.getCacheObject(key);
                List<LiveRoomGoodsBid> liveRoomGoodsBids = JSON.parseArray(cacheObject, LiveRoomGoodsBid.class);
                liveRoomGoodsBids.add(liveRoomGoodsBid);
                redisService.setCacheObject(key, JSON.toJSONString(liveRoomGoodsBids));
                //判断是否是最高价 若是则发送消息给主播
                if (liveRoomGoodsBids.size() == 1){
                    rocketMQTemplate.convertAndSend("live-auction-topic", liveRoomGoodsBid);
                }
            }else {
                List<LiveRoomGoodsBid> liveRoomGoodsBids = new ArrayList<>();
                liveRoomGoodsBids.add(liveRoomGoodsBid);
                redisService.setCacheObject(key, liveRoomGoodsBid);
            }



        }else if (liveAuctionSendPriceReq.getType()==1){
            //直接出价 需要根据主播设置的出价进行出
            LambdaQueryWrapper<CereShopMarketing> cereShopMarketingLambdaQueryWrapper = new LambdaQueryWrapper<>();
            cereShopMarketingLambdaQueryWrapper.eq(CereShopMarketing::getGoodsId,liveAuctionSendPriceReq.getProductId());
            cereShopMarketingLambdaQueryWrapper.eq(CereShopMarketing::getActivityType,3);
            CereShopMarketing one = cereShopMarketingMapper.selectOne(cereShopMarketingLambdaQueryWrapper);

            //获取当前直播间商品竞拍的最高价格
            BigDecimal topPrice = getTopPrice(liveAuctionSendPriceReq);

            if (one != null){
                BigDecimal addPrice = one.getAddPrice();
                if (topPrice.compareTo(BigDecimal.ZERO) == 0){
                    BigDecimal Price = one.getOriginalPrice().add(addPrice);
                    String key = LIVE_AUCTION_LIST + liveAuctionSendPriceReq.getLiveRoomId() + ":" + liveAuctionSendPriceReq.getProductId();
                    List<LiveRoomGoodsBid> liveAuctionSendPriceRes = new ArrayList<>();
                    LiveRoomGoodsBid liveRoomGoodsBid = new LiveRoomGoodsBid();
                    liveRoomGoodsBid.setBidAmount(Price);
                    liveRoomGoodsBid.setBidTime(DateUtil.toLocalDateTime(new Date()));
                    liveRoomGoodsBid.setUserId(LoginHelper.getUserId());
                    liveRoomGoodsBid.setRoomId(liveAuctionSendPriceReq.getLiveRoomId());
                    liveRoomGoodsBid.setGoodsId(liveAuctionSendPriceReq.getProductId());
                    liveRoomGoodsBid.setType(0L);
                    liveAuctionSendPriceRes.add(liveRoomGoodsBid);
                    redisService.setCacheObject(key, JSON.toJSONString(liveAuctionSendPriceRes));
                }


            }



        }

        //存入Redis 在直播结束后记录到数据库
        return R.ok();
    }

    /**
     * 获取当前直播间商品竞拍的最高价格
     * @param liveAuctionSendPriceReq
     * @return
     */
    private BigDecimal getTopPrice(LiveAuctionSendPriceReq liveAuctionSendPriceReq) {
        String key = LIVE_AUCTION_LIST + liveAuctionSendPriceReq.getLiveRoomId() + ":" + liveAuctionSendPriceReq.getProductId();
        String json = (String) redisService.getCacheObject(key);
        if (json == null){
            return BigDecimal.ZERO;
        }
        List<LiveRoomGoodsBid> liveAuctionSendPriceRes = JSON.parseArray(json, LiveRoomGoodsBid.class);
        if (CollUtil.isNotEmpty(liveAuctionSendPriceRes)){
            return liveAuctionSendPriceRes.stream()
                    .map(LiveRoomGoodsBid::getBidAmount)
                    .max(BigDecimal::compareTo)
                    .orElse(BigDecimal.ZERO);
        }
        return BigDecimal.ZERO;
    }

    @Override
    public R<?> snapProduct(SnapProductReq snapProductReq) {
        return null;
    }

    /**
     * 封装数据
     * @param records
     * @return
     */
    private List<RecommendRes> buildLiveRoomRes(List<CereBuyerLiveRoom> records) {
        List<RecommendRes> res = new ArrayList<>();
        for (CereBuyerLiveRoom record : records) {
            RecommendRes recommendRes = new RecommendRes();
            recommendRes.setAnchorHeadImage(record.getAnchorAvatar());
            recommendRes.setAnchorName(record.getAnchorName());
            recommendRes.setLiveCover(record.getCoverUrl());
            recommendRes.setLiveStatus(record.getStatus().toString());
            recommendRes.setLiveTitle(record.getTitle());
            recommendRes.setLiveTypeId(record.getCategoryId().toString());
            recommendRes.setLiveTypeName(record.getCategoryName());
            res.add(recommendRes);
        }
        return res;
    }

    private List<CereBuyerLiveRoom> getAdPromotionAnchor() {
        //开播之后 将广告位推广计入Redis 并设置时间计时
        List<String> allRecommendedLiveIds = redisService.getAllRecommendedLiveIds();
        if (allRecommendedLiveIds.isEmpty()){
            return List.of();
        }

        return lambdaQuery().in(CereBuyerLiveRoom::getId, allRecommendedLiveIds).list();
    }
}
